﻿using System.Threading.Channels;
using System.Threading.Tasks;

namespace JobDemo01
{
    internal class Program
    {
        static Task task;
        static Channel<string> channel = Channel.CreateUnbounded<string>();
        static MessageQueue<string> queue = new MessageQueue<string>();
        static ISubscription<string> sub;
        static ISubscription<string> sub1;

        static Func<string, bool> predicate;

        static bool Predicate(string s)
        {
            return s.Contains("Message");
        }

        static async Task Main(string[] args)
        {
            predicate = Predicate;
            int state = 2;
            int ooo = Interlocked.Exchange(ref state, 4);
            Console.WriteLine(state + $"    ooo:{ooo}");
            ooo = Interlocked.CompareExchange(ref state, 1, 1);

            Console.WriteLine(state + $"     ooo:{ooo}");
            ooo = Interlocked.CompareExchange(ref state, 1, 4);

            Console.WriteLine(state + $"     ooo:{ooo}");
            try
            {
                try
                {
                    task = channel.Reader.Completion;
                    channel.Writer.Complete(new Exception("Force exception")); // 添加这行
                    await task;
                    Console.WriteLine("11111111111111");
                }
                finally
                {
                    Console.WriteLine("22222222222222");

                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("11111111111111");
            }
            
            sub = queue.Subscribe(5);
            sub1 = queue.Subscribe();
            Job job = new Job();
            

            // 生产者线程
            Task t1 = Task.Run(() =>
            {
                for (int i = 0; i < 10; i++)
                {
                    queue.Send($"Message {i}");
                    Thread.Sleep(100);
                }
                queue.UnsubscribeAll(null); // 结束所有订阅
            });
            await t1 ;

            Console.WriteLine($"10条数据都发送完成...............");

            // 消费者线程
            _ = Task.Run(async () =>
            {
                try
                {
                    while (true)
                    {
                        string msg = await sub.Read();
                        string msg1 = await sub1.Read();
                        Console.WriteLine($"Consumed: {msg}                         {msg1}");
                    }
                }
                catch (ChannelClosedException)
                {
                    Console.WriteLine("Subscription closed.");
                }
            });


            Console.ReadKey();
        }
    }
}
